package database

import (
	"abdis/interface/database"
	"abdis/interface/redis"
	"abdis/redis/protocol"
	"strings"
)

func Watch(db *DB, conn redis.Connection, args [][]byte) redis.Reply {
	watching := conn.GetWatching()
	for _, bk := range args {
		key := string(bk)
		watching[key] = db.GetVersion(key)
	}
	return protocol.MakeOkReply()
}

// StartMulti 开启事务
func StartMulti(conn redis.Connection) redis.Reply {
	if conn.InMultiState() {
		return protocol.MakeErrReply("ERR MULTI calls can not be nested")
	}
	conn.SetMultiState(true)
	return protocol.MakeOkReply()
}

func DiscardMulti(conn redis.Connection) redis.Reply {
	if !conn.InMultiState() {
		return protocol.MakeErrReply("ERR not int multi")
	}
	conn.ClearQueuedCmds()
	conn.SetMultiState(false)
	return protocol.MakeOkReply()
}

// EnqueueCmd 将命令加入队列
func EnqueueCmd(conn redis.Connection, cmdLine [][]byte) redis.Reply {
	name := strings.ToLower(string(cmdLine[0]))
	cmd := getCmd(name)
	if cmd == nil {
		reply := protocol.MakeErrReply("ERR unknown command : " + name)
		conn.AddTxError(reply)
		return reply
	}
	if cmd.prepare == nil {
		reply := protocol.MakeErrReply("ERR command " + name + "cannot be used in multi")
		conn.AddTxError(reply)
		return reply
	}
	if !validateArity(cmd.arity, cmdLine) {
		reply := protocol.MakeArgNumErrReply(name)
		conn.AddTxError(reply)
		return reply
	}
	conn.EnqueueCmd(cmdLine)
	return protocol.MakeQueuedReply()
}

func execMulti(db *DB, conn redis.Connection) redis.Reply {
	if !conn.InMultiState() {
		return protocol.MakeErrReply("ERR not in multi")
	}
	defer conn.SetMultiState(false)

	if len(conn.GetTxErrors()) > 0 {
		return protocol.MakeErrReply("ERR EXECABORT Transaction discarded because of previous errors.")
	}

	cmdLines := conn.GetQueuedCmdLine()
	return db.ExecMulti(conn.GetWatching(), cmdLines)
}

func isWatchingChanged(db *DB, watch map[string]uint32) bool {
	for key, version := range watch {
		cur := db.GetVersion(key)
		if cur != version {
			return true
		}
	}
	return false
}

func (db *DB) ExecMulti(watching map[string]uint32, cmdLines []database.CmdLine) redis.Reply {
	// prepare
	writeKeys := make([]string, 0) // 可能有重复的键
	readKeys := make([]string, 0)
	for _, cmdLine := range cmdLines {
		name := strings.ToLower(string(cmdLine[0]))
		cmd := getCmd(name)
		prepare := cmd.prepare
		write, read := prepare(cmdLine[1:])
		writeKeys = append(writeKeys, write...)
		readKeys = append(readKeys, read...)
	}

	// watch
	watchingKeys := make([]string, 0, len(watching))
	for key := range watching {
		watchingKeys = append(watchingKeys, key)
	}
	readKeys = append(readKeys, watchingKeys...)

	db.RWLocks(writeKeys, readKeys)
	defer db.RWUnLocks(writeKeys, readKeys)

	if isWatchingChanged(db, watching) {
		// 乐观锁导致事务失效
		return protocol.MakeEmptyMultiBulkReply()
	}

	// execute
	results := make([]redis.Reply, 0, len(cmdLines))
	aborted := false
	undoCmdLines := make([][]database.CmdLine, 0, len(cmdLines))
	for _, cmdLine := range cmdLines {
		undoCmdLines = append(undoCmdLines, db.GetUndoLogs(cmdLine))
		result := db.execWithLock(cmdLine)
		if protocol.IsErrorReply(result) {
			aborted = true
			// 错误的命令后续不 rollback
			undoCmdLines = undoCmdLines[:len(undoCmdLines)-1]
			break
		}
		results = append(results, result)
	}

	if !aborted {
		// 成功执行
		db.addVersion(writeKeys...)
		return protocol.MakeMultiRawReply(results)
	}

	// 失败情况下 rollback
	size := len(undoCmdLines)
	for i := size - 1; i >= 0; i-- {
		// 反过来执行
		cur := undoCmdLines[i]
		if len(cur) == 0 {
			continue
		}
		for _, cmdLine := range cur {
			db.execWithLock(cmdLine)
		}
	}
	return protocol.MakeErrReply("ERR EXECABORT Transaction discarded because of previous errors.")
}

func (db *DB) GetUndoLogs(cmdLine [][]byte) []database.CmdLine {
	name := strings.ToLower(string(cmdLine[0]))
	cmd := getCmd(name)
	if cmd == nil {
		return nil
	}
	undo := cmd.undo
	if undo == nil {
		return nil
	}
	return undo(db, cmdLine[1:])
}
